package cmd

import (
	"bytes"
	"fmt"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/pingcap/tidb/config"
	"github.com/pingcap/tidb/store/tikv"
	"github.com/spf13/cobra"
)

var archiveCmd = &cobra.Command{
	Use:   "archive",
	Short: "Archive He3DB Xlog KV",
	Long:  "Welcome to use hbr for He3DB xlog archive",
	Run:   runArchive,
}

func init() {
	rootCmd.AddCommand(archiveCmd)
}

func runArchive(cmd *cobra.Command, args []string) {
	var sem = make(chan bool, concurrency)
	archiveStart := time.Now()
	access_key, _ := cmd.Flags().GetString("access_key")
	secret_key, _ := cmd.Flags().GetString("secret_key")
	endpoint, _ := cmd.Flags().GetString("endpoint")
	region, _ := cmd.Flags().GetString("region")
	bucket, _ := cmd.Flags().GetString("bucket")
	pd, _ := cmd.Flags().GetString("pd")
	backup_name, _ := cmd.Flags().GetString("name")
	archive_start_time_line, _ := cmd.Flags().GetString("archive_start_time_line")
	archive_start_lsn, _ := cmd.Flags().GetString("archive_start_lsn")

	if access_key == "" || secret_key == "" || endpoint == "" || region == "" || bucket == "" || pd == "" || backup_name == "" || archive_start_time_line == "" || archive_start_lsn == "" {
		fmt.Printf("PARAMETER ERROR!\n")
		return
	}

	client, err := tikv.NewRawKVClient([]string{pd}, config.Security{})
	if err != nil {
		fmt.Printf("Connect Tikv Error!\n%v\n", err)
		return
	}

	sess, err := session.NewSession(&aws.Config{
		Region:      aws.String(region),
		Endpoint:    aws.String(endpoint),
		Credentials: credentials.NewStaticCredentials(access_key, secret_key, ""),
		S3ForcePathStyle: aws.Bool(true),
	})
	if err != nil {
		fmt.Printf("Connect S3 Error!\n%v\n", err)
		return
	}
	s3_client := s3.New(sess)

	var filename string = ""
	wlCount := 0

	// archive wal kv
	fmt.Printf("archive wal kv!\n")
	for id := 0; id < 8; id++ {
		//06000000000000000100000000000000070000000000000000
		//因为加了个id字段，目前不能跨时间线备份
		retStartString := fmt.Sprintf("06%s000000000000000%d%s", archive_start_time_line, id, archive_start_lsn)
		//retEndString := fmt.Sprintf("06ffffffffffffffff000000000000000%dffffffffffffffff", id)
		retEndString := fmt.Sprintf("06%s000000000000000%dffffffffffffffff", archive_start_time_line, id)

		retStart := make([]byte, 25)
		retEnd := make([]byte, 25)
		index := 0
		for i := 0; i < len(retStartString); i += 2 {
			value, _ := strconv.ParseUint(retStartString[i:i+2], 16, 8)
			retStart[index] = byte(0xff & value)
			value, _ = strconv.ParseUint(retEndString[i:i+2], 16, 8)
			retEnd[index] = byte(0xff & value)
			index++
		}
		fmt.Printf("%x\n", retStart)
		fmt.Printf("%x\n", retEnd)

		limit := 10240

		for {
			keys, values, _ := client.Scan(retStart, retEnd, limit)
			for k, _ := range keys {
				fmt.Printf("%x\n", keys[k])
				filename = fmt.Sprintf("%x", keys[k])
				wg.Add(1)
				sem <- true
				go s3PutKV(s3_client, bucket, backup_name, filename, values[k], sem)

				if bytes.Compare(retStart, keys[k]) < 0 {
					retStart = keys[k]
				}
				wlCount++
			}
			if len(keys) < limit {
				break
			}
			wlCount--
		}
	}

	wg.Wait()
	client.Close()

	fmt.Printf("wal kv count:%v\n", wlCount)
	fmt.Println("backup time:", time.Since(archiveStart))
}
